#include <queue>
#include <list>
#include <rise/common/ExceptionTemplate.h>
#include <rise/common/exmacros.h>
#include <rise/threading/CriticalSection.h>
#include <staff/common/Value.h>
#include <staff/common/DataObject.h>
#include "Message.h"
#include "MessageQueue.h"

namespace smq
{
  struct MessageQueue::MessageQueueImpl
  {
    MessageQueueImpl():
      nQueueLimit(-1)
    {
    }

    typedef std::list<Message> MessageList;
    MessageList lsMessages;
    int nQueueLimit;
    rise::threading::CCriticalSection tLock;
  };


  MessageQueue::MessageQueue()
  {
    m_pImpl = new MessageQueueImpl;
  }

  MessageQueue::~MessageQueue()
  {
    if(m_pImpl)
    {
      delete m_pImpl;
      m_pImpl = NULL;
    }
  }

  MessageQueue::MessageQueue(const MessageQueue& rQueue)
  {
  }

  MessageQueue& MessageQueue::operator=(const MessageQueue& rQueue)
  {
    return *this;
  }

  void MessageQueue::AddMessage(const Message& rMsg)
  {
    rise::threading::CScopedCriticalSection(m_pImpl->tLock);
    m_pImpl->lsMessages.push_back(rMsg);
    if (m_pImpl->nQueueLimit != -1)
    {
      while (m_pImpl->lsMessages.size() > static_cast<unsigned>(m_pImpl->nQueueLimit))
      {
        rise::LogDebug2() << "removing message " << m_pImpl->lsMessages.front().sId << " from queue: limit for " << m_pImpl->nQueueLimit << " reaced";
        m_pImpl->lsMessages.pop_front();
      }
    }
  }

  void MessageQueue::RemoveMessage(const std::string& sMsgId)
  {
    rise::threading::CScopedCriticalSection(m_pImpl->tLock);
    for (MessageQueueImpl::MessageList::iterator itMsg = m_pImpl->lsMessages.begin();
        itMsg != m_pImpl->lsMessages.end(); ++itMsg)
    {
      if (itMsg->sId == sMsgId)
      {
        m_pImpl->lsMessages.erase(itMsg);
        break;
      }
    }
  }

  int MessageQueue::GetQueueLength()
  {
    return static_cast<int>(m_pImpl->lsMessages.size());
  }

  const Message& MessageQueue::GetLastMessage() const
  {
    MessageQueueImpl::MessageList::reverse_iterator itMsg = m_pImpl->lsMessages.rbegin();
    RISE_ASSERTS(itMsg != m_pImpl->lsMessages.rend(), "Can\'t get last message");
    return *itMsg;
  }

  void MessageQueue::GetMessageHeadersList(MessageHeaderList& rlsMessageHeaders) const
  {
    MessageHeader tHeader;
    rlsMessageHeaders.clear();
    for (MessageQueueImpl::MessageList::const_iterator itMsg = m_pImpl->lsMessages.begin();
      itMsg != m_pImpl->lsMessages.end(); ++itMsg)
    {
      tHeader.sId = itMsg->sId;
      tHeader.sPublisher = itMsg->sPublisher;
      tHeader.sQueue = itMsg->sQueue;
      rlsMessageHeaders.push_back(tHeader);
    }
  }

  const Message& MessageQueue::GetMessage(const std::string& sMsgId) const
  {
    MessageQueueImpl::MessageList::const_iterator itMsg = m_pImpl->lsMessages.begin();
    for (; itMsg != m_pImpl->lsMessages.end(); ++itMsg)
    {
      if (itMsg->sId == sMsgId)
      {
        break;
      }
    }

    RISE_ASSERTS(itMsg != m_pImpl->lsMessages.end(), "Can\'t get message: " + sMsgId);
    return *itMsg;
  }

  void MessageQueue::SetQueueLimit(int nLimit /*= 100*/)
  {
    m_pImpl->nQueueLimit = nLimit;
  }

}
